package com.xinxing.learning.crud.batch.transactional;

import lombok.extern.slf4j.Slf4j;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicReference;

@Slf4j
public class TransactionMultipartExecutor<T> {
    /**
     * 单个线程处理的数据量
     */
    private int singleCount;
    /**
     * 处理的总数据量
     */
    private int listSize;
    /**
     * 开启的线程数
     */
    private int runSize;
    /**
     * 操作的数据集
     */
    private List<T> list;

    /**
     * 计数器 拦截器
     */
    private CountDownLatch begin, end;
    /**
     * 线程池
     */
    private ExecutorService executorService;
    /**
     * 是否存在异常
     */
    private AtomicReference<Boolean> isError = new AtomicReference<>(false);
    /**
     * 回调函数
     */
    private CallBack callback;
    /**
     * 事务管理器
     */
    private PlatformTransactionManager transactionManager;

    public TransactionMultipartExecutor(int singleCount, List<T> list) {
        if (singleCount <= 0 || CollectionUtils.isEmpty(list)) {
            throw new RuntimeException("Illegal parameter");
        }
        transactionManager = ApplicationContextProvider.getApplicationContext().getBean(PlatformTransactionManager.class);
        this.singleCount = singleCount;
        this.list = list;
        this.listSize = list.size();
        this.runSize = this.listSize % this.singleCount == 0 ? this.listSize / this.singleCount : this.listSize / this.singleCount + 1;
    }

    public void excute() throws InterruptedException {
        try {
            this.executorService = Executors.newFixedThreadPool(this.runSize);
            this.begin = new CountDownLatch(1);
            this.end = new CountDownLatch(runSize);
            int startIndex = 0;
            int endIndex = 0;
            List<T> newList = null;
            for (int i = 0; i < runSize; i++) {
                if (i < (runSize - 1)) {
                    startIndex = i * singleCount;
                    endIndex = (i + 1) * singleCount;
                    newList = this.list.subList(startIndex, endIndex);
                } else {
                    startIndex = i * singleCount;
                    endIndex = this.listSize;
                    newList = list.subList(startIndex, endIndex);
                }

                BatchInsertThread<T> batchInsertThread = new BatchInsertThread<T>(newList, begin, end) {

                    @Override
                    public void method(List<T> list) {
                        DefaultTransactionDefinition defaultTransactionDefinition = new DefaultTransactionDefinition();
                        defaultTransactionDefinition.setPropagationBehavior(DefaultTransactionDefinition.PROPAGATION_REQUIRES_NEW);
                        TransactionStatus transactionStatus = transactionManager.getTransaction(defaultTransactionDefinition);
                        try {
                            callback.method(list);
                            log.warn("多线程事务批量操作执行成功,线程名:{},操作成功数量:{}", Thread.currentThread().getName(), list.size());
                        } catch (Exception e) {
                            // 接收异常,处理异常
                            isError.set(true);
                            //e.printStackTrace();
                            log.error("多线程事务批量操作抛错,线程名:{},操作失败数量:{},报错信息:{},{}", Thread.currentThread().getName(), list.size(), e.toString(), e);
                        }
                        end.countDown();

                        try {
                            //等待所有线程任务完成，监控是否有异常，有则统一回滚
                            //log.warn("等待所有任务执行完成,当前时间:{},当前end计数:{}", LocalDateTime.now(), end.getCount());
                            end.await();
                            //log.warn("完成所有任务,当前时间:{},当前end计数:{}", LocalDateTime.now(), end.getCount());
                            if (isError.get()) {
                                // 事务回滚
                                transactionManager.rollback(transactionStatus);
                            } else {
                                //事务提交
                                transactionManager.commit(transactionStatus);
                            }
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                };
                executorService.submit(batchInsertThread);
            }
            // 通过上面的for循环 已经将数据拆分到多个线程中 并且将线程任务提交到了线程池中 现成任务中有begin.await()阻塞住
            // begin.countDown()执行后线程池的每个线程将会开始执行
            begin.countDown();
            end.await();

        } finally {
            // 任务全部执行完毕
            executorService.shutdown();
        }


    }

    public void setCallback(CallBack callback) {
        this.callback = callback;
    }
}
